package com.bauer.springboot.starter.context;

import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.bauer.springboot.starter.annotation.AliListener;
import com.bauer.springboot.starter.inerface.AliMQListener;
import java.lang.reflect.ParameterizedType;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.lang.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.core.annotation.AnnotationUtils;

/**
 * @description:
 * @author: weirui
 * @create: 2020-01-17 15:35
 **/
public class AliMQSubUtils {

    private static Logger logger = LoggerFactory.getLogger(AliMQSubUtils.class);

    public static Map<Subscription, MessageListener> getSubscription(ApplicationContext applicationContext) {
        String[] listeners = applicationContext.getBeanNamesForAnnotation(AliListener.class);
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<>();
        if (ArrayUtils.isEmpty(listeners)) {
            logger.warn("no aliMQListeners init");
            return subscriptionTable;
        }

        for (String listenerName : listeners) {
            AliMQListener listener = (AliMQListener) applicationContext.getBean(listenerName);
            AliListener aliListener = AnnotationUtils.findAnnotation(listener.getClass(), AliListener.class);
            String topic = aliListener.topic();
            String tag = aliListener.tag();
            Subscription subscription = new Subscription();
            subscription.setTopic(topic);
            subscription.setExpression(tag);
            subscriptionTable.put(subscription, new MessageListener() {
                @Override
                public Action consume(Message message, ConsumeContext consumeContext) {
                    try {
                        Class genericClass = (Class) ((ParameterizedType) getClass().getGenericSuperclass())
                            .getActualTypeArguments()[0];

                        listener.consume(JSON.parseObject(message.getBody(), genericClass));
                    } catch (Exception ex) {
                        return Action.ReconsumeLater;
                    }
                    return Action.CommitMessage;
                }
            });
        }

        return subscriptionTable;
    }


}
